package kafka

import (
	"github.com/Shopify/sarama"
	"strings"
	"log"
	"time"
)

type Kafka struct{
	brokerList []string
	topicName string
	syncSender sarama.SyncProducer
}

func NewKafka(brokerList ,topic string)(k *Kafka,err error){
	k = new(Kafka)
	k.brokerList = strings.Split(brokerList,",")
	k.topicName = topic
	k.syncSender,err = sarama.NewSyncProducer(k.brokerList,nil)
	return k,err
}

func (k Kafka)Send(data string)error{
	msg := sarama.ProducerMessage{
		Topic:k.topicName,
		Value: sarama.StringEncoder(data),
		Timestamp: time.Now(),
	}
	_, _, err := k.syncSender.SendMessage(&msg)

	if err != nil {
		log.Printf("FAILED to send message: %s\n", err)
		return err
	}

	//partition, offset, err := k.syncSender.SendMessage(&msg)
	//if err != nil {
	//	log.Printf("FAILED to send message: %s\n", err)
	//	return err
	//} else {
	//	log.Printf("> message sent to partition %d at offset %d\n", partition, offset)
	//}
	return nil
}
func (k Kafka)Close(){
	k.syncSender.Close()
}

//func client()  {
//	c ,err:= sarama.NewClient([]string{"10.1.10.48:9092"},nil)
//	c.RefreshMetadata()
//	c.Closed()
//}